package com.zt.bigdata.flink.stream

import java.util

import com.zt.bigdata.template.spark.redis.RedisTemplate
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.{TimeWindow, Window}
import org.apache.flink.util.Collector
import org.apache.flink.streaming.api.scala._
import redis.clients.jedis.Jedis

/**
  * @ClassName ApplyDemo
  * @Description
  * @Author zhangtonghy
  * @Date 2019-07-16 11:10
  * @Copyright: 版权所有 (C) zt zt.
  * @注意 ：本内容仅限于zt内部传阅，禁止外泄以及用于其他的商业目的
  **/
object ApplyDemo {

  class MyWindowFunction2 extends WindowFunction[(String, Int), String, String, TimeWindow] {
    def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[String]): Unit = {
      var count = 0L
      for (in <- input) {
        count = count + 1
      }
      out.collect(s"Window $window count: $count")
    }
  }

  class MyWindowFunction3 extends WindowFunction[(String, Int), Unit, String, TimeWindow] {

    var jedis: Jedis = null;

    def open(): Unit = {
      jedis = RedisTemplate.get(0)
    }

    def close(): Unit = {
      jedis.close()
    }

    def process(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[Unit]): Unit = {
      val map = new util.HashMap[String, String]()
      input.foreach {
        x => {
          map.put(x._1, x._2.toString)
        }
      }
      jedis.hmset("adb", map)
    }


    def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[Unit]): Unit = {
      open()
      process(key, window, input, out)
      close()
    }
  }


  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val input: DataStream[String] = env.socketTextStream("localhost", 999)
    val word = input
      .map {
        x => {
          val pair = x.split("\\s+")
          (pair(0), pair(1).toInt)
        }
      }

    //    word.keyBy(_._1)
    //      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    //      .apply(new MyWindowFunction2()).print()
    //
    //    word.keyBy(_._1)
    //      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    //      .apply((key, window, in, out: Collector[(String, Long, Long, Int)]) => {
    //        out.collect((key, window.getStart, window.getEnd, in.map(_._2).sum))
    //      }).print()

    RedisTemplate.init("localhost:6379")

    word.keyBy(_._1)
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
      .apply(new MyWindowFunction3())

    env.execute()
  }
}
